package com.atguigu.edu.realtime.common.base;


import com.atguigu.edu.realtime.common.util.SQLUtil;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public abstract class BaseSQLApp {
    public void start(int port,int paral,String ck){
        //1 基本环境
        Configuration conf = new Configuration();
        conf.set(RestOptions.PORT,port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(paral);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 2.检查点相关的设置
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));

        //2 检查点设置
//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//       CheckpointConfig checkpointConfig = env.getCheckpointConfig();
//        checkpointConfig.setCheckpointTimeout(60000L);
//        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        checkpointConfig.setMinPauseBetweenCheckpoints(2000L);
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
//        env.setStateBackend(new HashMapStateBackend());
//        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/ck/"+ck);
//        System.setProperty("HADOOP_USER_NAME","atguigu");

        //3 业务处理
       handle(env,tableEnv);
    }

    public abstract void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) ;

    public  void readOdsDS(StreamTableEnvironment tableEnv,String groupId) {
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "  `database` string,\n" +
                "  `table` string,\n" +
                "  `type` string,\n" +
                "  `data` map<string,string>,\n" +
                "  `old` map<string,string>,\n" +
                "  ts bigint,\n" +
                "  pt as proctime(),\n" +
                "  et as TO_TIMESTAMP_LTZ(ts, 0),\n" +
                "  watermark for et as et\n" +
                ") "+ SQLUtil.getKafkaDDL("topic_db",groupId));
    }



    public void readBaseDic(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("CREATE TABLE base_dic (\n" +
                " dic_code string,\n" +
                " info ROW<dic_name string>,\n" +
                " PRIMARY KEY (dic_code) NOT ENFORCED\n" +
                ") "+SQLUtil.getHBaseDDL("dim_base_dic"));
    }
}
